package com.bootdo.clouddo.redis;

import com.alibaba.fastjson.support.spring.GenericFastJsonRedisSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import redis.clients.util.SafeEncoder;

import java.util.*;
import java.util.concurrent.TimeUnit;

import static java.util.stream.Collectors.toList;

@Component
public class CommonRedisService {

	protected Logger logger = LoggerFactory.getLogger(getClass());
	
	@Autowired
 	private StringRedisTemplate stringRedisTemplate;

	@Autowired
	private RedisTemplate<String, Object> redisTemplate;

	@Value("${redis.scope}")
	private String redis_scope;

	private GenericFastJsonRedisSerializer jsonRedisSerializer = new GenericFastJsonRedisSerializer();

	public RedisTemplate<String, Object> getRedisTemplate() {
		return redisTemplate;
	}

	public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
		this.redisTemplate = redisTemplate;
	}

	/**
	 * 从Redis中取出数据
	 * @param key
	 * @return
	 */
	
	  public Object get(String key) {
	        final String keyf = getAppKey(key);
	        Object object = null;
	        try {
				object = redisTemplate.execute(new RedisCallback<Object>() {
					@Override
					public Object doInRedis(RedisConnection connection)
							throws DataAccessException {
						  byte[] key = keyf.getBytes();
				            byte[] value = connection.get(key);
				            if (value == null) {
				                return null;
				            }
				            return jsonRedisSerializer.deserialize(value);
					}
				});
			} catch (Exception e) {
	        	e.printStackTrace();
				logger.error("error", e);//默认只打印日志，避免其他业务方法忘记捕获运行时异常时不走数据库
			}
	        return object;
	    }
	  
	  /**
	   * 将数据放入Redis中
	   * @param key
	   * @param value
	   * @param expire 单位秒
	   */
	  	public void put(String key, Object value, Long expire) {  
	        final String fKey = getAppKey(key);
	        final Object fValue = value;  
	        final Long fe = expire;
			redisTemplate.execute(new RedisCallback<Long>() {  
			    public Long doInRedis(RedisConnection connection)  
			            throws DataAccessException {
			        byte[] keyb = fKey.getBytes();  
			        byte[] valueb = jsonRedisSerializer.serialize(fValue);
			        if(fe != null && fe > 0){
			        	connection.setEx(keyb, fe, valueb);  
			        }
			        else{
			        	connection.set(keyb, valueb);
			        }
			        return 1L;  
			    }  
			}); 
	    }

	    @Async
		public void asyncPut(String key, Object value, Long expire) {
			put(key, value, expire);
		}

		/**
         * 将数据放入redis，如果此key不存在的话；否则失败
         * @param key
         * @param value
         */
	    public void putNX(String key, Object value) {  
	        final String fKey = getAppKey(key);  
	        final Object fValue = value;  
			redisTemplate.execute(new RedisCallback<Long>() {  
			    public Long doInRedis(RedisConnection connection)  
			            throws DataAccessException {
			        byte[] keyb = fKey.getBytes();  
			        byte[] valueb = jsonRedisSerializer.serialize(fValue);
			    
			        connection.setNX(keyb, valueb);  
			        
			        return 1L;  
			    }  
			});
	    }

		public boolean setNX(final String key, final String value) {
			Object obj = null;
			try {
				obj = redisTemplate.execute(new RedisCallback<Object>() {
					@Override
					public Object doInRedis(RedisConnection connection) throws DataAccessException {
						StringRedisSerializer serializer = new StringRedisSerializer();
						Boolean success = connection.setNX(serializer.serialize(key), serializer.serialize(value));
						connection.close();
						return success;
					}
				});
			} catch (Exception e) {
				logger.error("setNX redis error, key : {}", key);
			}
			return obj != null ? (Boolean) obj : false;
		}

		/**
		 * 多服务器集群，使用下面的方法，代替System.currentTimeMillis()，获取redis时间，避免多服务的时间不一致问题！！！
		 *
		 * @return
		 */
		public long currtTimeForRedis() {
			Object obj = redisTemplate.execute(new RedisCallback<Object>() {
				@Override
				public Object doInRedis(RedisConnection redisConnection) throws DataAccessException {
					return redisConnection.time();
				}
			});
			return obj == null ? -1 : Long.parseLong(obj.toString());
		}

		public boolean expire(final String key, final Long expireTime) {
			Object obj = null;
			try {
				obj =  redisTemplate.execute(new RedisCallback<Boolean>() {
					@Override
					public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
						StringRedisSerializer serializer = new StringRedisSerializer();
						boolean bool = connection.expire(serializer.serialize(key), expireTime);
						connection.close();
						return bool;
					}
				});
				return (Boolean)obj;
			} catch (Exception e) {
				logger.error("expire redis error, key : {}", key);
			}
			return false;
		}
	    
	    /**
	     * 将key绑定的value从redis中删除
	     * @param key
	     */
	   
	    public void remove(String key){
	    	final String fKey = getAppKey(key);  
	        try {
				redisTemplate.execute(new RedisCallback<Long>() {  
				    public Long doInRedis(RedisConnection connection)  
				            throws DataAccessException {
				        byte[] keyb = fKey.getBytes();  
				        connection.del(keyb);  
				        return 1L;  
				    }  
				});
			} catch (Exception e) {
				logger.error("error", e);//默认只打印日志，避免其他业务方法忘记捕获运行时异常时不走数据库
			}  
	    }
	    public boolean existObjectKey(String key){
			return redisTemplate.hasKey(getAppKey(key)); 
		}

		public boolean existsKey(String key) {
			return redisTemplate.execute(
					(RedisConnection conn) -> conn.exists(getAppKey(key).getBytes()));
	    }
		/**
		 * 往listName列表右边插入值value
		 * @param key
		 * @param value
		 */
		public void setList(String key,String value){
			stringRedisTemplate.opsForList().rightPush(getAppKey(key), value);
		}
		/**
		 * 列表listName中的左边第一个对象出栈
		 * @param key
		 * @return
		 */
		public String getList(String key){
			return stringRedisTemplate.opsForList().leftPop(getAppKey(key));
		}
		public Long getListLength(String key){
			return stringRedisTemplate.opsForList().size(getAppKey(key));
		}
		public void expireListKey(String key,int timeout,TimeUnit unit){
			stringRedisTemplate.expire(getAppKey(key), timeout, unit);

		}
		public long getListKeyExpire(String key){
			return stringRedisTemplate.getExpire(getAppKey(key));
		}
		
		 
		public boolean existListKey(String key){
			return stringRedisTemplate.hasKey(getAppKey(key)); 
		}
		public void removeListKey(String key){
			stringRedisTemplate.delete(getAppKey(key));

		}
		private String getAppKey(String key){
			return redis_scope+key;
		}
		
		public Set<String> keys(String pattern){
			final String fKey = getAppKey(pattern);
	        try {
				Set<byte[]> resultSet = redisTemplate.execute(new RedisCallback<Set<byte[]>>() {  
				    public Set<byte[]> doInRedis(RedisConnection connection)  
				            throws DataAccessException {
				        byte[] keyb = fKey.getBytes();  
				        return connection.keys(keyb);  
				    }  
				});
				HashSet<String> strSet = new HashSet<String>();
				for (byte[] bs : resultSet) {
					strSet.add(new String(bs).substring(redis_scope.length()));
				}
				return strSet;

				
			} catch (Exception e) {
				logger.error("error", e);//默认只打印日志，避免其他业务方法忘记捕获运行时异常时不走数据库
			}  
	        return null;
		}
		
		public void removeKeys(String pattern){
			final String fKey = getAppKey(pattern);
			try {
				redisTemplate.execute(new RedisCallback<Void>() {  
					public Void doInRedis(RedisConnection connection)  
							throws DataAccessException {
						byte[] keyb = fKey.getBytes();  
						Set<byte[]> keys = connection.keys(keyb);  
						for (byte[] bs : keys) {
							connection.del(bs);
						}
						return null;
					}  
				});
				
			} catch (Exception e) {
				logger.error("error", e);//默认只打印日志，避免其他业务方法忘记捕获运行时异常时不走数据库
			}  
		}
		
		
		/**
		 * 镜像层_加载
		 * @param key 
		 * @param time
		 * @param parameter 搜索参数
		 * @param rslt	爬来的数据
		 */
		@SuppressWarnings("unchecked")
		public void mirroringCache(String key,Long time,String parameter,Map<String,Object> rslt){
			key = getAppKey(key);
			Map<String,Object> map = null;
			if(this.get(key)!=null){
				map = (Map<String, Object>) this.get(key);
				map.put(parameter, rslt);
				this.put(key, map, time);
			}else{
				map = new HashMap<String, Object>();
				map.put(parameter, rslt);
				this.put(key, map, time);
			}
		}
		
		/**
		 * 镜像层_读取
		 * @param key
		 * @param parameter 搜索参数
		 * @return
		 */
		@SuppressWarnings("unchecked")
		public Object mirroringCache(String key,String parameter){
			key = getAppKey(key);
			Map<String, Object> rstl = new HashMap<String, Object>();
			if(this.get(key)==null || ((Map<String, Object>) this.get(key)).get(parameter)==null){
				rstl.put("desc", "fail");
				rstl.put("status", "500");
				return rstl;	
			}else{
				return ((Map<String, Object>) this.get(key)).get(parameter);
			}
		}
		
		/**
		 * 镜像层_清除
		 * @param keyArray KEY数组
		 */
		public void mirroringClear(String[] keyArray){
			for(int i=0;i<keyArray.length;i++){
				this.remove(getAppKey(keyArray[i]));
			}
		}
		/**
		 * 往listName列表右边插入值value
		 * @param key
		 * @param value
		 */
		public void setListObject(String key, Object value){
			redisTemplate.execute(
					(RedisConnection conn) -> conn.rPush(getAppKey(key).getBytes(), jsonRedisSerializer.serialize(value)));
		}
		/**
		 * 列表listName中的左边第一个对象出栈
		 * @param key
		 * @return
		 */
		public Object getListObject(String key){
			return redisTemplate.execute(
					(RedisConnection conn) -> jsonRedisSerializer.deserialize(conn.lPop(getAppKey(key).getBytes())));
			//return redisTemplate.opsForList().leftPop(getAppKey(key));
		}
		
		
		
		/**
		 * ZADD
		 * @return
		 */
		public boolean zAdd(String key, double score, Object value) {
			return redisTemplate.execute(
					(RedisConnection conn) -> conn.zAdd(getAppKey(key).getBytes(), score, jsonRedisSerializer.serialize(value)));
		}
		
		/**
		 * ZREM
		 * 
		 * @return
		 */
		public boolean zRem(String key, Object value) {
			return redisTemplate.execute(
					(RedisConnection conn) -> conn.zRem(getAppKey(key).getBytes(), jsonRedisSerializer.serialize(value))>0);
		}
		
		/**
		 * ZCARD
		 * 
		 * @return
		 */
		public long zCard(String key) {
			return redisTemplate.execute(
					(RedisConnection conn) -> conn.zCard(getAppKey(key).getBytes()));
		}

		public long zRemRangeByScore(String key, double min, double max) {
			return redisTemplate.execute(
					(RedisConnection conn) -> conn.zRemRangeByScore(getAppKey(key).getBytes(), min, max));
		}

		public long zRemRangeByScore(String key, double score) {
			return redisTemplate.execute(
					(RedisConnection conn) -> conn.zRemRangeByScore(getAppKey(key).getBytes(), score, score));
		}



		/**
		 * ZRANGEBYSCORE
		 * 
		 * @return
		 */
		@SuppressWarnings("unchecked")
		public <O> List<O> zRangeByScore(String key, double min, double max, long offset, long count) {
			Set<byte[]> result = redisTemplate.execute(
					(RedisConnection conn) -> conn.zRangeByScore(getAppKey(key).getBytes(), min, max, offset, count));
			return result.stream().map(jsonRedisSerializer::deserialize).map(obj -> (O) obj).collect(toList());
		}
		
		/**
		 * ZREVRANGEBYSCORE
		 * 
		 * @return
		 */
		@SuppressWarnings("unchecked")
		public <O> List<O> zRevRangeByScore(String key, double min, double max, long offset, long count) {
			Set<byte[]> result = redisTemplate.execute(
					(RedisConnection conn) -> conn.zRevRangeByScore(getAppKey(key).getBytes(), min, max, offset, count));
			return result.stream().map(jsonRedisSerializer::deserialize).map(obj -> (O) obj).collect(toList());
		}
		
		/**
		 * HSET
		 * 
		 * @return
		 */
		public boolean hSet(String key, String field, Object value) {
			return redisTemplate.execute(
					(RedisConnection conn) -> conn.hSet(getAppKey(key).getBytes(), field.getBytes(), jsonRedisSerializer.serialize(value)));
		}
		
		/**
		 * HSETNX
		 * 
		 * @return
		 */
		public boolean hSetNX(String key, String field, Object value) {
			return redisTemplate.execute(
					(RedisConnection conn) -> conn.hSetNX(getAppKey(key).getBytes(), field.getBytes(), jsonRedisSerializer.serialize(value)));
		}
		
		/**
		 * HGET
		 * 
		 * @return
		 */
		@SuppressWarnings("unchecked")
		public <O> Optional<O> hGet(String key, String field) {
			byte[] result = redisTemplate.execute(
					(RedisConnection conn) -> conn.hGet(getAppKey(key).getBytes(), field.getBytes()));
			return Optional.ofNullable((O) jsonRedisSerializer.deserialize(result));
		}
		
		/**
		 * HMGET
		 * 
		 * @return
		 */
		@SuppressWarnings("unchecked")
		public <O> List<O> hMGet(String key, String... fields) {
			byte[][] bFields = Arrays.stream(fields).map(String::getBytes).toArray(byte[][]::new);
			List<byte[]> result = redisTemplate.execute(
					(RedisConnection conn) -> conn.hMGet(getAppKey(key).getBytes(), bFields));
			return result.stream().map(jsonRedisSerializer::deserialize).map(obj -> (O) obj).collect(toList());
		}
		
		/**
		 * HDEL
		 * 
		 * @return
		 */
		public boolean hDel(String key, String field) {
			return redisTemplate.execute(
					(RedisConnection conn) -> conn.hDel(getAppKey(key).getBytes(), field.getBytes()) >0);
		}
		/**
		 * HKEYS
		 * @param key
		 * @return
		 */
		public Set<String> hKeys(String key){
			return redisTemplate.execute(new RedisCallback<Set<String>>() {
				@Override
				public Set<String> doInRedis(RedisConnection connection) throws DataAccessException {
					Set<byte[]> setResult=connection.hKeys(getAppKey(key).getBytes());
					Set<String> setFinal=new HashSet<String>();
					for(byte[] bytes:setResult){
						setFinal.add(new String(bytes));
					}
					return setFinal;
				}
			});
		}
		
		/**
		 * 从Redis中取出数据
		 * 
		 * @return
		 */
		@SuppressWarnings("unchecked")
		public <O> Optional<O> getAsOptional(String key) {
			Object obj = redisTemplate
					.execute((RedisConnection conn) -> jsonRedisSerializer.deserialize(conn.get(getAppKey(key).getBytes())));
			return Optional.ofNullable((O) obj);
		}

	public boolean lock(String key, String value, Long lockExpireTimeOut,
						 Long lockWaitTimeOut) {
		if(lockExpireTimeOut != null && lockExpireTimeOut > 0)
			return false;
		try {
			Long deadTimeLine = System.currentTimeMillis() + lockWaitTimeOut;
			for (;;) {
				final Object fValue = value;
				Long result = redisTemplate.execute(new RedisCallback<Long>() {
					public Long doInRedis(RedisConnection connection)
							throws DataAccessException {
						Object obj = connection.execute("set", new byte[][]{
								jsonRedisSerializer.serialize( getAppKey(key)), jsonRedisSerializer.serialize(value),
								jsonRedisSerializer.serialize("NX"), jsonRedisSerializer.serialize("EX"),
								SafeEncoder.encode(String.valueOf(lockExpireTimeOut))

						});
						if (obj != null && "OK".equals(obj.toString())) {
							return 1L;
						}
						return 0L;
					}
				});
				if(result == 1)
					return true;

				lockWaitTimeOut = deadTimeLine - System.currentTimeMillis();

				if (lockWaitTimeOut <= 0L) {
					return false;
				}
			}
		} catch (Exception ex) {
			logger.info("lock error");
		}
		return false;
	}

	public boolean unlock(String key, String value) {
		try {
			String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
			final String fKey = getAppKey(key);
			Long result = redisTemplate.execute(new RedisCallback<Long>() {
				public Long doInRedis(RedisConnection connection)
						throws DataAccessException {
					Long result =connection.eval(luaScript.getBytes(), ReturnType.INTEGER, 1, getAppKey(key).getBytes(), value.getBytes());
					if (result == 1) {
						return 1L;
					}
					return 0L;
				}
			});
			if(result == 1)
				return true;
		} catch (Exception ex) {
			logger.info("unlock error");
		}
		return false;

	}

}
